package com.zhao.biz.etl

import com.zhao.entity.Log
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

import java.io.File
import scala.util.Random

/**
 * Description: 将日志文件转换成Parquet格式的数据<br/>
 * Copyright (c) ，2021 ， 赵 <br/>
 * A wet person does not fear the rain. <br/>
 * Date： 2021/1/12 17:18
 *
 * @author 柒柒
 * @version : 1.0
 */

object log2Parquet {
  def main(args: Array[String]): Unit = {

    //接收参数
    val inputPath = "a_data/2018-10-01_06_p1_invalid.1475274123982.log"
    val outputPath = "a_data/outputpath"
    val inputPath2 = "a_data/longAndlat.txt"
    val Array(input, output, input2) = Array(inputPath, outputPath, inputPath2)

    //val warehouseLocation = new File("spark-warehouse").getAbsolutePath


    //1.SparkSession
    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName)
      .master("local[*]")
      //设置序列化的技术(使用kryo)
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      //设置snappy压缩的格式
      .config("spark.sql.parquet.compression.code", "snappy")
      .config("spark.sql.warehouse.dir","D:/DaiMa/IDEA/Code_Project/a_data")
      .getOrCreate()

    val sc: SparkContext = spark.sparkContext

    Logger.getRootLogger.setLevel(Level.ERROR)

    //2.将维度信息置于广播变量中(若是日志文件中不包含经纬度信息,随机造数据)
    //给出经纬度信息,便于后续的计算
    val longAndLastRDD: Array[(String, String)] = sc.textFile(input2)
      .map(_.split("\\s+"))
      .map(arr => (arr(0).trim, arr(1).trim))
      .collect()

    val bcLongAndLast: Broadcast[Array[(String, String)]] = sc.broadcast(longAndLastRDD)

    //3.注册实体类(本质,那个实体类的实例使用到了Kryo序列化技术)
    sc.getConf.registerKryoClasses(Array(classOf[Log]))

    //4.加载日志数据到内存中,映射为DataFrame
    import spark.implicits._
    val df: DataFrame = sc.textFile(inputPath)
      .map(perLine => perLine.split(","))
      //数据清洗过滤(将Log中长度!=85的元素筛选掉)
      .filter(_.length == 85)
      .map(Log(_))
      //分析数据,若是经纬度没有数据的话,造数据,但是,真实项目中,关于经纬度的信息,不能随意造数据
      .map(log => {
        val _long = log._long.trim
        val lat = log.lat.trim
        if (_long.isEmpty || lat.isEmpty || "0".equals(_long) || "0".equals(lat)) {
          val arr = bcLongAndLast.value
          //获得经纬度数组汇总随机索引值
          val randomIndex = Random.nextInt(arr.length)
          val randomEle: (String, String) = arr(randomIndex)
          log._long = randomEle._1
          log.lat = randomEle._2
        }
        log
      }).toDF()

    //将清洗后的数据落地到目的地
    df.write.partitionBy("provincename","cityname")
      .mode(SaveMode.Overwrite)
      .parquet(outputPath)
  }
}

















